package secsmarts;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

public class LoadFileToHbase {
	public static class MapperClass extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put> {
		// 列名
		public static final String[] COLUMNS = { "time", "url", "ip", "country", "code", "type" };

		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			String[] cloumnVals = value.toString().split("\t");
			String url=cloumnVals[1], ip=cloumnVals[2],host=gethost(url);
			String[] rowkeys = new String[] { url,ip,host};
			for (String rowkey : rowkeys) {
				Put put = new Put(rowkey.getBytes());
				for (int i = 0; i < cloumnVals.length; i++) {
					put.add("info".getBytes(), COLUMNS[i].getBytes(), cloumnVals[i].getBytes());
				}
				context.write(new ImmutableBytesWritable(rowkey.getBytes()), put);
			}
		}

		private String gethost(String url) {
			// TODO Auto-generated method stub
			if(url!=null&&url.startsWith("http://"))
			{
				String tmp=url.substring(7);
				return url.substring(0,7+tmp.indexOf("/"));
			}
			return null;
		}
	}

	static String tableName = "bingmurls";

	public static void main(String[] args) {
		try {
			 
			createTable();
			Configuration conf = HBaseConfiguration.create();
			conf.set("fs.defaultFS", "hdfs://secsmarts");
			conf.set("mapreduce.framework.name", "yarn");
			// conf.set("mapred.job.tracker", "ubuntu:9001");
			conf.set("hbase.zookeeper.quorum", "slave01.secsmarts.com,master.secsmarts.com,slave02.secsmarts.com");
			conf.set("hbase.zookeeper.property.clientPort", "2181");
			conf.set("zookeeper.znode.parent", "/hbase-unsecure");
			Job job = new Job(conf, "file2hbase");
			job.setJarByClass(LoadFileToHbase.class);
			job.setMapperClass(MapperClass.class);
			job.setNumReduceTasks(0);
			job.setOutputFormatClass(TableOutputFormat.class);
			job.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, tableName);
			FileInputFormat.addInputPath(job, new Path(args[0]));
			System.out.println(job.waitForCompletion(true) ? 0 : 1);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private static void createTable() throws IOException {
		// TODO Auto-generated method stub
		 
	    Configuration conf = HBaseConfiguration.create();
	    conf.set("hbase.zookeeper.quorum", "slave01.secsmarts.com,master.secsmarts.com,slave02.secsmarts.com");
	    conf.set("zookeeper.znode.parent", "/hbase-unsecure");
	    HBaseAdmin admin = new HBaseAdmin(conf);
	    if(admin.tableExists(tableName)){
	    	return;   
	    }
	    System.out.println("table not exists!creating.......");
        admin.disableTable(tableName);
        admin.deleteTable(tableName);
	    HTableDescriptor htd = new HTableDescriptor(tableName);
	    HColumnDescriptor tcd = new HColumnDescriptor("info");
	    htd.addFamily(tcd);//创建列族
	    admin.createTable(htd);//创建表
	}
}
